package org.redis.cluster.redis.transport;

import java.util.Collection;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.redis.cluster.RedisCluster;
import org.redis.cluster.config.ClientConfig;
import org.redis.cluster.extension.ExtensionLoader;
import org.redis.cluster.loadbalance.LoadBalance;
import org.redis.cluster.plugin.Plugin;
import org.redis.cluster.redis.RedisCommand;
import org.redis.cluster.redis.transport.reply.RedisByteReply;
import org.redis.cluster.redis.transport.reply.RedisErrorReply;
import org.redis.cluster.redis.transport.reply.RedisTimeoutReply;
import org.redis.cluster.transport.client.ClientPoolFactory;
import org.redis.cluster.transport.client.Client;
import org.redis.cluster.transport.client.ClientManager;
import org.redis.cluster.transport.client.ClientPool;
import org.redis.cluster.transport.exception.TimeoutException;
import org.redis.cluster.transport.session.Session;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class DefaultRedisServerProcessor implements RedisServerProcessor,Plugin {
	
	private final static String CLIENT_MANAGER="clientManager";
	
	private final static String LOAD_BALANCE="loadBalance";
	
	protected final Logger logger=LoggerFactory.getLogger(getClass());

	private RedisCluster redisCluster;
	private ExtensionLoader extensionLoader;
	private ClientManager clientManager;
	private LoadBalance loadBalance;
	
	@Override
	public void init(RedisCluster redisCluster) {
		this.redisCluster=redisCluster;
		this.extensionLoader=redisCluster.getExtensionLoader();
		this.loadBalance=initLoadBalance();
		this.clientManager=initClientManager();
	}
	
	protected LoadBalance initLoadBalance() {
		return extensionLoader.load(LOAD_BALANCE,LoadBalance.class);
	}
	
	protected ClientManager initClientManager() {
		return extensionLoader.load(CLIENT_MANAGER,ClientManager.class);
	}
	
	@Override
	public void destroy() {
	}
	
	protected ClientPool getClientPool() {
		return ClientPoolFactory.getClientPool();
	}
	
	@Override
	public RedisReply process(Session session, RedisCommand command) {
		ClientConfig targetClientConfig=command.getTarget();
		if(targetClientConfig==null) {
			Collection<ClientConfig> clientConfigs=clientManager.getConfigs();
			
			if(CollectionUtils.isEmpty(clientConfigs)) {
				return new RedisTimeoutReply();
//				return new RedisErrorReply("no nodes available");
			}
			
			targetClientConfig=doSelectTargetClientConfig(clientConfigs,command);
			
			if(targetClientConfig==null) {
				return new RedisTimeoutReply();
//				return new RedisErrorReply("no nodes available");
			}
		}
		
		command.setTarget(targetClientConfig);
		
		if(ArrayUtils.isEmpty(command.getSource())) {
			return null;
		}
		
		try {
			return doInvoke(targetClientConfig,command);
		}
		catch(TimeoutException e) {
			if(logger.isDebugEnabled()) {
				logger.debug("调用超时  - {} - {}",targetClientConfig,command);
			}
			
			return new RedisTimeoutReply();
		}
	}
	
	protected RedisReply doInvoke(ClientConfig clientConfig,RedisCommand command) throws TimeoutException {
		if(logger.isDebugEnabled()) {
			logger.debug("准备调用后端redis - {} - {}",clientConfig,command);
		}
		
		ClientPool clientPool=getClientPool();
		Client client=null;
		
		try {
			client=clientPool.getClient(clientConfig);
		}
		catch(Exception e) {
			if(logger.isErrorEnabled()) {
				logger.error("",e);
			}
			
			return new RedisErrorReply(e.getMessage());
		}
		
		if(client==null) {
			return new RedisTimeoutReply();
//			return new RedisErrorReply("no connection available");
		}
		
		if(logger.isDebugEnabled()) {
			logger.debug("调用后端redis - {} - {}",clientConfig,command);
		}
		
		try {
			byte response[]=client.send(command.getSource());
			
			if(logger.isDebugEnabled()) {
				logger.debug("后端redis返回结果 - {} - {}",ArrayUtils.toString(response),command);
			}
			
			return new RedisByteReply(response);
		}
		finally {
			clientPool.returnClient(clientConfig, client);
		}
	}
	
	protected ClientConfig doSelectTargetClientConfig(Collection<ClientConfig> clientConfigs,RedisCommand command) {
		if(command.getParameterCount()==1) {
			return (ClientConfig) CollectionUtils.get(clientConfigs, 0);
		}
		
		return loadBalance.select(clientConfigs,command);
	}

}